﻿
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;
using System;
using System.Collections.Generic;
using System.Net.Sockets;
using System.Reflection;
using System.Text;
using System.Threading;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.DependencyInjection;
using System.Collections.Concurrent;
using EventBus.Common;

namespace EventBus.RabbitMQ
{
    public class RabbitMqSubscriber : IRabbitMqSubscriber
    {
        ConcurrentDictionary<string, int> ConcurrentExceptionDic = new ConcurrentDictionary<string, int>();
        private readonly ILogger<RabbitMqSubscriber> _logger;
        private readonly IConnectionChannel _connectionChannel;
        public RabbitMqSubscriber(IConnectionChannel connectionChannel, ILogger<RabbitMqSubscriber> logger)
        {
            _connectionChannel = connectionChannel;
            _logger = logger;
        }
        public RabbitMqSubscriber() { }

        /// <summary>
        /// 订阅Command
        /// </summary>
        /// <typeparam name="TCommand">命令</typeparam>
        /// <param name="serviceProvider">依赖注入的实例</param>
        /// <param name="queueName">队列名称</param>
        /// <param name="deadLetter">死信队列</param>
        public void SubscriberCommand<TCommand>(IServiceProvider serviceProvider, string queueName, bool deadLetter = false) where TCommand : ICommand
        {
            var commandHandlerImpl = serviceProvider.GetService<ICommandHandler<TCommand>>();
            var channel = _connectionChannel.GetSubscriberChannel();
            if (deadLetter)
            {
                var dlxExchangeName = $"dlxExchange{queueName}";
                var dlxQueueName = $"dlxQueue{queueName}";
                var dlxRoutingKey = $"dlxRoutingKey{queueName}";
                // 创建死信交换器和队列
                channel.ExchangeDeclare(dlxExchangeName, "topic", true, false);
                channel.QueueDeclare(dlxQueueName, true, false, false);
                channel.QueueBind(dlxQueueName, dlxExchangeName, dlxRoutingKey);
                var args = new Dictionary<string, object>
                {
                    { "x-dead-letter-exchange", dlxExchangeName},
                    { "x-dead-letter-routing-key",dlxRoutingKey}
                    //{ "x-message-ttl",60000} //设置消息的存活时间，即过期时间
                };
                //声明队列
                channel.QueueDeclare(queue: queueName, durable: true, false, false, args);
            }
            else
            {
                channel.QueueDeclare(queue: queueName, durable: true, false, false);
            }

            //绑定交换机队列
            channel.QueueBind(queue: queueName, exchange: _connectionChannel.ExchangeName, routingKey: queueName);

            var consumer = new EventingBasicConsumer(channel);
            channel.BasicQos(0, 1000, false);
            consumer.Shutdown += OnConsumerShutdown;
            consumer.Registered += OnConsumerRegistered;
            consumer.Unregistered += OnConsumerUnregistered;
            consumer.ConsumerCancelled += OnConsumerConsumerCancelled;
            consumer.Received += (model, ea) =>
             {
                 try
                 {
                     var body = ea.Body;
                     var message = Encoding.Default.GetString(body);
                     var command = SwifterJsonSerializer.DeserializeObject<TCommand>(message);
                     commandHandlerImpl.HandleAsync(command).GetAwaiter().GetResult();
                     channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                 }
                 catch (Exception ex)
                 {
                     var command = SwifterJsonSerializer.DeserializeObject<TCommand>(Encoding.Default.GetString(ea.Body));
                     ConcurrentExceptionDic.TryGetValue(command.CommandId, out int exceptionCount);
                     if (deadLetter)
                     {
                         if (exceptionCount >= 5)
                         {
                             channel.BasicNack(ea.DeliveryTag, false, false);
                         }
                         else
                         {
                             ConcurrentExceptionDic.TryRemove(command.CommandId, out int exceptionCountValue);
                             ConcurrentExceptionDic.TryAdd(command.CommandId, exceptionCountValue + 1);
                             channel.BasicNack(ea.DeliveryTag, false, true);
                         }
                     }
                     else
                     {
                         channel.BasicNack(ea.DeliveryTag, false, true);
                     }
                 }
             };
            channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
        }

        /// <summary>
        /// 订阅事件
        /// </summary>
        /// <typeparam name="TEvent">命令</typeparam>
        /// <param name="serviceProvider">依赖注入的实例</param>
        /// <param name="queueName">队列名称</param>
        /// <param name="deadLetter">死信队列</param>
        public void SubscriberEvent<TEvent>(IServiceProvider serviceProvider, string queueName, bool deadLetter = false) where TEvent : IEvent
        {
            var commandHandlerImpl = serviceProvider.GetService<IEventHandler<TEvent>>();
            var channel = _connectionChannel.GetSubscriberChannel();
            if (deadLetter)
            {
                var dlxExchangeName = $"dlxExchange{queueName}";
                var dlxQueueName = $"dlxQueue{queueName}";
                var dlxRoutingKey = $"dlxRoutingKey{queueName}";
                // 创建死信交换器和队列
                channel.ExchangeDeclare(dlxExchangeName, "topic", true, false);
                channel.QueueDeclare(dlxQueueName, true, false, false);
                channel.QueueBind(dlxQueueName, dlxExchangeName, dlxRoutingKey);
                var args = new Dictionary<string, object>
                {
                    { "x-dead-letter-exchange", dlxExchangeName},
                    { "x-dead-letter-routing-key",dlxRoutingKey}
                    //{ "x-message-ttl",60000} //设置消息的存活时间，即过期时间
                };
                //声明队列
                channel.QueueDeclare(queue: queueName, durable: true, false, false, args);
            }
            else
            {
                channel.QueueDeclare(queue: queueName, durable: true, false, false);
            }

            //绑定交换机队列
            channel.QueueBind(queue: queueName, exchange: _connectionChannel.ExchangeName, routingKey: queueName);

            var consumer = new EventingBasicConsumer(channel);
            channel.BasicQos(0, 1000, false);
            consumer.Shutdown += OnConsumerShutdown;
            consumer.Registered += OnConsumerRegistered;
            consumer.Unregistered += OnConsumerUnregistered;
            consumer.ConsumerCancelled += OnConsumerConsumerCancelled;
            consumer.Received += (model, ea) =>
            {
                try
                {
                    var body = ea.Body;
                    var message = Encoding.Default.GetString(body);
                    var command = SwifterJsonSerializer.DeserializeObject<TEvent>(message);
                    commandHandlerImpl.HandleAsync(command).GetAwaiter().GetResult();
                    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                }
                catch (Exception ex)
                {
                    var command = SwifterJsonSerializer.DeserializeObject<TEvent>(Encoding.Default.GetString(ea.Body));
                    ConcurrentExceptionDic.TryGetValue(command.AggregateRootId, out int exceptionCount);
                    if (deadLetter)
                    {
                        if (exceptionCount >= 5)
                        {
                            channel.BasicNack(ea.DeliveryTag, false, false);
                        }
                        else
                        {
                            ConcurrentExceptionDic.TryRemove(command.AggregateRootId, out int exceptionCountValue);
                            ConcurrentExceptionDic.TryAdd(command.AggregateRootId, exceptionCountValue + 1);
                            channel.BasicNack(ea.DeliveryTag, false, true);
                        }
                    }
                    else
                    {
                        channel.BasicNack(ea.DeliveryTag, false, true);
                    }
                }
            };
            channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
        }

        /// <summary>
        /// 取消消费者
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnConsumerConsumerCancelled(object sender, ConsumerEventArgs e)
        {
            _logger.LogInformation("取消消费者", e.ConsumerTag);
        }

        /// <summary>
        /// 消费者取消注册
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnConsumerUnregistered(object sender, ConsumerEventArgs e)
        {
            _logger.LogInformation("消费者取消注册", e.ConsumerTag);
        }

        /// <summary>
        /// 消费者注册
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnConsumerRegistered(object sender, ConsumerEventArgs e)
        {
            _logger.LogInformation("消费者注册", e.ConsumerTag);
        }

        /// <summary>
        /// 消费者宕机
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnConsumerShutdown(object sender, ShutdownEventArgs e)
        {
            _logger.LogInformation("消费者宕机", e.ReplyText);
        }

    }

}
